package com.easydev.mq.rabbit;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.easydev.mq.MessageTemplate;
import com.easydev.mq.converter.JsonMessageConverter;
import com.easydev.mq.converter.MessageConverter;
import com.easydev.mq.exception.MessageException;
import com.easydev.mq.rabbit.pool.RabbitChannelPoolTool;
import com.rabbitmq.client.Channel;

/**
 * <Description> rabbitMQ消息发送类，用于发送消息，默认以JSON形式发送<br>
 * sendMessage:P2P形式发送，发送至指定队列，若无队列则消息丢失 ；对应RABBIT 默认转发器<br>
 * sendTopicMessage：简单广播消息发送，用于基于订阅/发布模式（Pub/Sub）对应RABBIT FANOUT转发器<br>
 * sendAdvanceMessage:根据转发器类型发送消息：FANOUT、DIRECT、TPOIC、HEADERS<br>
 * 
 * @author lx<br>
 * @version 1.0<br>
 * @CreateDate 2016年3月15日 <br>
 */
public class RabbitMessageTemplate implements MessageTemplate {
    private final static Logger LOGGER = LoggerFactory.getLogger(RabbitMessageTemplate.class);

    private RabbitChannelPoolTool channelPool;

    private MessageConverter messageConverter;

    public RabbitMessageTemplate() throws MessageException {
        this.channelPool = RabbitChannelPoolTool.getInstance();
        this.messageConverter = new JsonMessageConverter();
    }

    @Override
    public void sendMessage(String queneName, Object message) throws Exception {
        sendMessage(StringUtils.EMPTY, queneName, message, "");
    }

    public void sendMessage(String queneName, Object message, String virtualHost) throws Exception {
        sendMessage(StringUtils.EMPTY, queneName, message, virtualHost);
    }

    @Override
    public void sendTopicMessage(String topicName, Object message) throws Exception {
        sendMessage(topicName, StringUtils.EMPTY, message, "");
    }

    public void sendTopicMessage(String topicName, Object message, String virtualHost) throws Exception {
        sendMessage(topicName, StringUtils.EMPTY, message, virtualHost);
    }

    /**
     * Description: 根据指定路由及规则发布消息<br>
     * 
     * @author lx<br>
     * @param exchange rabbit消息转发器
     * @param routingKey 路由规则
     * @param message 消息内容
     * @throws MessageException <br>
     */
    public void sendAdvanceMessage(String exchange, String routingKey, Object message, String virtualHost)
            throws Exception {
        sendMessage(exchange, routingKey, message, virtualHost);
    }

    public void sendAdvanceMessage(String exchange, String routingKey, Object message) throws Exception {
        sendMessage(exchange, routingKey, message, "");
    }

    @Override
    public Object createQueue(String queneName) throws Exception {
        return createQueue(queneName, Boolean.FALSE);
    }

    public Object createQueue(String queneName, String virtualHost) throws Exception {
        return createQueue(queneName, Boolean.FALSE, virtualHost);
    }

    @Override
    public Object createQueue(String queneName, boolean durable) throws Exception {
        return declareQueue(queneName, durable, StringUtils.EMPTY, StringUtils.EMPTY, StringUtils.EMPTY);
    }

    public Object createQueue(String queneName, boolean durable, String virtualHost) throws Exception {
        return declareQueue(queneName, durable, StringUtils.EMPTY, StringUtils.EMPTY, virtualHost);
    }

    /**
     * Description: 创建队列，如果声明了转发器则绑定转发器<br>
     * 
     * @author lx<br>
     * @param queneName 队列名称
     * @param durable 是否持久化 false 否 true 是
     * @param exchange 转发器，如果非空，则绑定当前queue至转发器
     * @param routingKey 如果是DIRECT、TPOIC、HEADERS类型是的路由规则
     * @return
     * @throws MessageException <br>
     */
    public String createQueue(String queneName, boolean durable, String exchange, String routingKey) throws Exception {
        return declareQueue(queneName, durable, exchange, routingKey, StringUtils.EMPTY);
    }

    public String createQueue(String queneName, boolean durable, String exchange, String routingKey, String virtualHost)
            throws Exception {
        return declareQueue(queneName, durable, exchange, routingKey, virtualHost);
    }

    @Override
    public void createTopic(String topicName) throws Exception {
        createExchange(topicName, "fanout");
    }

    public void createTopic(String topicName, String virtualHost) throws Exception {
        createExchange(topicName, "fanout", virtualHost);
    }

    /**
     * Description: 根据类型创建转发器 基于默认virtualHost<br>
     * 
     * @author lx<br>
     * @param topicName 转发器名称
     * @param type 转发器类型：FANOUT、DIRECT、TPOIC、HEADERS
     * @throws MessageException <br>
     */
    public void createExchange(String topicName, String type) throws Exception {
        createExchange(topicName, topicName, "");
    }

    /**
     * Description: 根据类型创建转发器<br>
     * 
     * @author lx<br>
     * @param topicName 转发器名称
     * @param type 转发器类型：FANOUT、DIRECT、TPOIC、HEADERS
     * @param virtualHost 虚拟目录
     * @throws MessageException <br>
     */
    public void createExchange(String topicName, String type, String virtualHost) throws Exception {
        Channel channel = null;
        try {
            channel = channelPool.borrowObject(virtualHost);
            channel.exchangeDeclare(topicName, type);
        } catch (Exception e) {
            LOGGER.error("RabbitMessageTemplate:::createExchange>>>create topic error ", e);
            throw new MessageException("create exchange error ", e);
        } finally {
            if (null != channel) {
                channelPool.returnObject(virtualHost, channel);
            }
        }
    }

    @Override
    public void deleteQueue(String queneName) throws Exception {
        deleteQueue(queneName, "");
    }

    public void deleteQueue(String queneName, String virtualHost) throws Exception {
        Channel channel = null;
        try {
            channel = channelPool.borrowObject(virtualHost);
            channel.queueDelete(queneName);
        } catch (Exception e) {
            LOGGER.error("RabbitMessageTemplate:::deleteQueue>>>delete queue error ", e);
            throw new MessageException("delete queue error ", e);
        } finally {
            if (null != channel) {
                channelPool.returnObject(virtualHost, channel);
            }
        }
    }

    /**
     * Description: 删除指定转发器<br>
     * 
     * @author lx<br>
     * @param topicName
     * @throws MessageException <br>
     */
    @Override
    public void deleteTopic(String topicName) throws Exception {
        deleteTopic(topicName, "");
    }

    public void deleteTopic(String topicName, String virtualHost) throws Exception {
        Channel channel = null;
        try {
            channel = channelPool.borrowObject(virtualHost);
            channel.exchangeDelete(topicName);
        } catch (Exception e) {
            LOGGER.error("RabbitMessageTemplate:::deleteTopic>>>delete exchange error ", e);
            throw new MessageException("delete exchange error ", e);
        } finally {
            if (null != channel) {
                channelPool.returnObject(virtualHost, channel);
            }
        }
    }

    private void sendMessage(String exchange, String routingKey, Object message, String virtualHost) throws Exception {
        Channel channel = null;
        try {
            channel = channelPool.borrowObject(virtualHost);
            // 如果Rabbit中没有名为topicName的exchange，将抛出exchange not found异常
            channel.basicPublish(exchange, routingKey, null, messageConverter.converterToByte(message));
        } catch (Exception e) {
            LOGGER.error("RabbitMessageTemplate:::sendMessage>>>send topic message error ", e);
            throw new MessageException("send topic message error ", e);
        } finally {
            if (null != channel) {
                channelPool.returnObject(virtualHost, channel);
            }
        }
    }

    private String declareQueue(String queneName, boolean durable, String exchange, String routingKey,
            String virtualHost) throws Exception {
        Channel channel = null;
        try {
            channel = channelPool.borrowObject(virtualHost);
            if (StringUtils.isBlank(queneName)) {
                queneName = StringUtils.EMPTY;
            }
            if (StringUtils.isBlank(routingKey)) {
                routingKey = StringUtils.EMPTY;
            }
            queneName = channel.queueDeclare(queneName, durable, false, false, null).getQueue();
            if (StringUtils.isNotBlank(exchange)) {
                channel.queueBind(queneName, exchange, routingKey);// TODO 是否需要考虑转发器不存在
            }
            return queneName;
        } catch (Exception e) {
            LOGGER.error("RabbitMessageTemplate:::declareQueue>>>declare Queue error ", e);
            throw new MessageException("declare Queue error", e);
        } finally {
            if (null != channel) {
                channelPool.returnObject(virtualHost, channel);
            }
        }
    }

}
